#include "StdAfx.h"
#include "AsyncSendQueueForSelectModel.h"

IAsyncSendQueue* AsyncSendQueueForSelectModel::CreateComponent(DISCONNECT_FUNCTOR disconnectFunctor)
{
	return new CAsyncSendQueueForSelectModel(disconnectFunctor);
}

CAsyncSendQueueForSelectModel::CAsyncSendQueueForSelectModel(DISCONNECT_FUNCTOR disconnectFunctor)
	: m_disconnectFunctor(disconnectFunctor)
{
}


CAsyncSendQueueForSelectModel::~CAsyncSendQueueForSelectModel(void)
{
}


bool CAsyncSendQueueForSelectModel::coreSend( SOCKET socket, IStreamBuffer& stream )
{
	int iResult = send( socket, stream.GetRaw()+stream.GetReadOffset(), stream.GetLength()-stream.GetReadOffset(), 0 );
	if (iResult == SOCKET_ERROR) 
	{
		if (WSAEWOULDBLOCK != WSAGetLastError())
		{
			m_disconnectFunctor(socket);
			return false;
		}		
	}

	stream.Read(iResult);
	return true;
}

bool CAsyncSendQueueForSelectModel::Send( SOCKET socket, IStreamBuffer& stream )
{
	Concurrency::critical_section::scoped_lock s(m_sendLock);

	bool isSendNow = true;

	auto it = m_sendQueueBySocket.find(socket);
	if (it != m_sendQueueBySocket.end())
	{
		isSendNow = it->second.empty();		
	}

	if (isSendNow)
	{
		coreSend(socket, stream);
	}

	if ( stream.GetReadOffset() < stream.GetLength())
	{
		m_sendQueueBySocket[socket].push_back(&stream);		
	}
	else
	{
		stream.Release();
	}

	return true;
}

bool CAsyncSendQueueForSelectModel::Send( SOCKET socket, size_t bytesSent )
{
	Concurrency::critical_section::scoped_lock s(m_sendLock);

	auto it = m_sendQueueBySocket.find(socket);
	if (it == m_sendQueueBySocket.end())
	{
		return false;
	}

	if (it->second.empty())
	{
		return false;
	}

	IStreamBuffer* pSteamBuffer = *it->second.begin();
	coreSend(socket, *pSteamBuffer);

	if (pSteamBuffer->GetLength() == pSteamBuffer->GetReadOffset())
	{
		it->second.pop_front();
		pSteamBuffer->Release();
	}

	return true;
}

void CAsyncSendQueueForSelectModel::Release()
{
	delete this;
}
